package com.atguigu.gmall.realtime.util

import java.util

import com.alibaba.fastjson.JSON
import com.alibaba.fastjson.serializer.SerializeConfig
import org.apache.http.HttpHost
import org.elasticsearch.action.bulk.BulkRequest
import org.elasticsearch.action.index.IndexRequest
import org.elasticsearch.action.search.{SearchRequest, SearchResponse}
import org.elasticsearch.client.{RequestOptions, RestClient, RestClientBuilder, RestHighLevelClient}
import org.elasticsearch.common.xcontent.XContentType
import org.elasticsearch.index.query.{BoolQueryBuilder, MatchQueryBuilder, RangeQueryBuilder}
import org.elasticsearch.search.SearchHit
import org.elasticsearch.search.aggregations.{AggregationBuilders, Aggregations}
import org.elasticsearch.search.aggregations.bucket.terms.{ParsedDoubleTerms, ParsedStringTerms, Terms, TermsAggregationBuilder}
import org.elasticsearch.search.aggregations.metrics.{AvgAggregationBuilder, ParsedAvg}
import org.elasticsearch.search.builder.SearchSourceBuilder
import org.elasticsearch.search.fetch.subphase.highlight.HighlightBuilder

object MyEsUtil {



    private var  restHighLevelClient:RestHighLevelClient= getClientNew


    def getClientNew:RestHighLevelClient={
      if(restHighLevelClient==null){
        val restClientBuilder: RestClientBuilder = RestClient.builder(new HttpHost("hdp1",9200))
        restHighLevelClient= new RestHighLevelClient(restClientBuilder)
      }
      restHighLevelClient
    }


    // 写入
    def save(): Unit ={
      val json=s"""{
        |"id":"55555",
        |"movie_name":"zhang555555555555"
        }""".stripMargin
        val indexRequest = new IndexRequest()
      indexRequest.index("movie_test0819_2022-01-17")
      indexRequest.source(json,XContentType.JSON)
      restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT)

    }

  //批量提交  batch  //幂等性 ？
  def  bulkSave(sourceList:List[(String,AnyRef)],indexName:String): Unit ={

    val bulkRequest = new BulkRequest(indexName)

    for ((id,source) <- sourceList ) {
      val indexRequest = new IndexRequest()
      indexRequest.source(JSON.toJSONString(source,new SerializeConfig(true)),XContentType.JSON)
      indexRequest.id(id)
      bulkRequest.add(indexRequest)
    }
    restHighLevelClient.bulk(bulkRequest,RequestOptions.DEFAULT)
    println("es存储" + sourceList.size + "条数据")
  }

  def main(args: Array[String]): Unit = {
    //save()
//    val movies = List(("999",Movie("999","movie999")),("888",Movie("888","movie888")))
//    bulkSave(movies,"movie_test0819_2022-01-17")
//    println("保存成功")


   // search()

    searchAgg()
  }


  //查询
  def search(): Unit ={

    val searchSourceBuilder = new SearchSourceBuilder()
    //查询
    val boolQueryBuilder = new BoolQueryBuilder()
    boolQueryBuilder.must(new MatchQueryBuilder("name","red sea"))
    boolQueryBuilder.filter(new RangeQueryBuilder("doubanScore").gt(8))
    searchSourceBuilder.query(boolQueryBuilder)
    //分页
    searchSourceBuilder.size(20)
    searchSourceBuilder.from(0)
    //高亮
    val highlightBuilder: HighlightBuilder = new HighlightBuilder().field("name")
    searchSourceBuilder.highlighter(highlightBuilder)


    val searchRequest = new SearchRequest()
    searchRequest.indices("movie_index0819")
    searchRequest.source(searchSourceBuilder)
    val searchResponse: SearchResponse = restHighLevelClient.search(searchRequest,RequestOptions.DEFAULT)


    val hits: Array[SearchHit] = searchResponse.getHits.getHits
    for (hit <- hits ) {
      val map: util.Map[String, AnyRef] = hit.getSourceAsMap
      println(map)
    }

  }


  def searchAgg(): Unit ={

    val searchSourceBuilder = new SearchSourceBuilder()
    searchSourceBuilder.size(0)

// 聚合
    val aggregationBuilder: TermsAggregationBuilder = AggregationBuilders.terms("group_actor").field("actorList.name.keyword").size(10)
    val avgAggs: AvgAggregationBuilder = AggregationBuilders.avg("avg_score").field("doubanScore")
    aggregationBuilder.subAggregation(avgAggs)
    searchSourceBuilder.aggregation(aggregationBuilder)


    val searchRequest = new SearchRequest()
    searchRequest.indices("movie_index0819")
    searchRequest.source(searchSourceBuilder)

    val searchResponse: SearchResponse = restHighLevelClient.search(searchRequest,RequestOptions.DEFAULT)

    val aggregations: Aggregations = searchResponse.getAggregations
    val group_actor: ParsedStringTerms = aggregations.get[ParsedStringTerms]("group_actor")
    val buckets: util.List[_ <: Terms.Bucket] = group_actor.getBuckets
    import  scala.collection.JavaConverters._
    for ( bucket <-buckets.asScala  ) {
      println(bucket.getKey)
      println(bucket.getDocCount)

      val avg_score: ParsedAvg = bucket.getAggregations.get[ParsedAvg]("avg_score")
      println(avg_score.getValue)

    }

  }



  case class Movie(id:String,movie_name:String)







}
